package SparkStreamingKafka

import kafka.serializer.StringDecoder
import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka.KafkaUtils
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * SparkStreaming接收Kafka推送过来的消息
  */
object SparkKafka {

  def main(args: Array[String]): Unit = {
    //构建conf ssc 对象
    val conf = new SparkConf().setAppName("Kafka_director").setMaster("local[2]")
    val ssc = new StreamingContext(conf, Seconds(5))
    //设置数据检查点进行累计统计单词
    //ssc.checkpoint("hdfs://192.168.11.31:9000/checkpoint")
    ssc.checkpoint("D:/wordcount")
    //kafka 需要Zookeeper  需要消费者组
    val topics = Set("SparkKafka")
    //                                     broker的原信息                                  ip地址以及端口号
    val kafkaPrams = Map[String, String]("metadata.broker.list" -> "192.168.11.31:9092")
    //                                          数据的输入了类型    数据的解码类型
    val data = KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaPrams, topics)
    val updateFunc = (curVal: Seq[Int], preVal: Option[Int]) => {
      //进行数据统计当前值加上之前的值
      val total = curVal.sum
      //最初的值应该是0
      var previous = preVal.getOrElse(0)
      //Some 代表最终的但会值
      Some(total + previous)
    }
    //统计结果
    val result = data.map(_._2).flatMap(_.split(" ")).map(word => (word, 1)).updateStateByKey(updateFunc).print()
    //启动程序
    ssc.start()
    ssc.awaitTermination()
  }

}
